package com.example.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 发送批量消息
 *
 * 如果您每次只发送不超过【4MB】的消息，则很容易使用批处理
 */
@Service
public class BathProducer {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    public void bathProducer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("bath-producer");
        producer.setNamesrvAddr(nameServer);
        producer.start();
        String topic = "broker-bath";
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
        try {
            SendResult send = producer.send(messages);
            System.out.println(send);
        } catch (Exception e) {
            e.printStackTrace();
            //处理error
        }finally {
            producer.shutdown();
        }
    }

    /**
     * 复杂度只有当你发送大批量时才会增长，你可能不确定它是否超过了大小限制（4MB）。
     * 这时候你最好把你的消息列表分割一下
     */
    public void bathProducer2() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("bath-producer2");
        producer.setNamesrvAddr(nameServer);
        producer.start();
        List<Message> list = buildList();
        ListSplitter listSplitter = new ListSplitter(list);
        while (listSplitter.hasNext()) {
            try {
                List<Message> listItem = listSplitter.next();
                producer.send(listItem);
            } catch (Exception e) {
                e.printStackTrace();
                //处理error
            }finally {
                producer.shutdown();
            }
        }
    }

    private class ListSplitter implements Iterator<List<Message>> {
        private final int SIZE_LIMIT = 1024 * 1024 * 4;
        private final List<Message> messages;
        private int currIndex;

        public ListSplitter(List<Message> messages) {
            this.messages = messages;
        }

        @Override
        public boolean hasNext() {
            return currIndex < messages.size();
        }

        @Override
        public List<Message> next() {
            int startIndex = getStartIndex();
            int nextIndex = startIndex;
            int totalSize = 0;
            for (; nextIndex < messages.size(); nextIndex++) {
                Message message = messages.get(nextIndex);
                int tmpSize = calcMessageSize(message);
                if (tmpSize + totalSize > SIZE_LIMIT) {
                    break;
                } else {
                    totalSize += tmpSize;
                }
            }
            List<Message> subList = messages.subList(startIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }

        private int getStartIndex() {
            Message currMessage = messages.get(currIndex);
            int tmpSize = calcMessageSize(currMessage);
            while (tmpSize > SIZE_LIMIT) {
                currIndex += 1;
                Message message = messages.get(currIndex);
                tmpSize = calcMessageSize(message);
            }
            return currIndex;
        }

        private int calcMessageSize(Message message) {
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
            return tmpSize;
        }
    }

    private List<Message> buildList(){
        String topic = "broker-bath2";
        List<Message> list = new ArrayList<>();
        for (int i = 0; i <10000; i++) {
            Message message = new Message(topic,("批量消息：hello-"+i).getBytes());
            list.add(message);
        }
        return list;
    }
}
